package com.joker.demo.core.executor.smart;


import cn.hutool.core.date.DateUtil;
import com.joker.demo.core.domain.dto.JobExecutorResultDTO;
import com.joker.demo.core.domain.dto.TaskExecuteRequestDTO;
import com.joker.demo.core.domain.dto.TaskExecuteResponseDTO;
import com.joker.demo.core.domain.dto.ThreadPoolMonitorDTO;
import com.joker.demo.core.domain.enums.JobRunStatusEnum;
import com.joker.demo.core.executor.JobExecutor;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Date;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.FutureTask;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicLong;

import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.Executors;

/**
 * 任务执行管理器
 * 使用线程池和FutureTask来控制任务执行，支持手动终止
 * 使用Netty时间轮处理延迟任务
 *
 * @author : feixiang.li
 * @since : 2025-07-24 16:57
 */
public class TaskExecutionManager {

    private static final Logger log = LoggerFactory.getLogger(TaskExecutionManager.class);

    /**
     * 任务执行线程池
     */
    private final ThreadPoolExecutor taskExecutor;

    /**
     * 任务映射表：jobId -> FutureTask
     */
    private final ConcurrentHashMap<Long, FutureTask<TaskExecuteResponseDTO>> taskMap = new ConcurrentHashMap<>();

    /**
     * 任务ID生成器
     */
    private final AtomicLong taskIdGenerator = new AtomicLong(0);

    /**
     * 任务执行器服务
     */
    private final JobExecutor jobExecutor;

    /**
     * Netty时间轮，用于处理延迟任务
     * 时间轮配置：
     * - tickDuration: 1秒，时间轮的刻度间隔
     * - ticksPerWheel: 512个刻度，时间轮的大小
     */
    private final HashedWheelTimer delayTimer;

    /**
     * 延迟任务映射表：jobId -> Timeout
     * 用于管理延迟任务，支持取消延迟任务
     */
    private final ConcurrentHashMap<Long, Timeout> delayTaskMap = new ConcurrentHashMap<>();

    /**
     * 线程池监控器
     */
    private ThreadPoolMonitor threadPoolMonitor;

    /**
     * 监控数据发送调度器
     */
    private ScheduledExecutorService monitorScheduler;

    public TaskExecutionManager(JobExecutor jobExecutor) {
        this.jobExecutor = jobExecutor;

        // 创建线程池
        this.taskExecutor = new ThreadPoolExecutor(
                // 核心线程数
                10,
                // 最大线程数
                50,
                // 空闲线程存活时间
                60L,
                // 时间单位
                TimeUnit.SECONDS,
                // 工作队列
                new LinkedBlockingQueue<>(1000), new ThreadFactory() {
            private final AtomicLong threadId = new AtomicLong(0);

            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r, "task-executor-" + threadId.incrementAndGet());
                thread.setDaemon(false);
                return thread;
            }
        },
                // 拒绝策略
                new ThreadPoolExecutor.CallerRunsPolicy());

        // 创建时间轮
        this.delayTimer = new HashedWheelTimer(
                new ThreadFactory() {
                    private final AtomicLong threadId = new AtomicLong(0);

                    @Override
                    public Thread newThread(Runnable r) {
                        Thread thread = new Thread(r, "delay-timer-" + threadId.incrementAndGet());
                        thread.setDaemon(true);
                        return thread;
                    }
                },
                1, // tickDuration: 1秒
                TimeUnit.SECONDS,
                512 // ticksPerWheel: 512个刻度
        );

        log.info("任务执行管理器初始化完成，线程池配置：核心线程数={}, 最大线程数={}", taskExecutor.getCorePoolSize(), taskExecutor.getMaximumPoolSize());
    }

    /**
     * 设置线程池监控器
     *
     * @param threadPoolMonitor 线程池监控器
     */
    public void setThreadPoolMonitor(ThreadPoolMonitor threadPoolMonitor) {
        this.threadPoolMonitor = threadPoolMonitor;
    }

    /**
     * 启动监控数据发送
     *
     * @param registerKey 执行器标识
     * @param appName 应用名称
     * @param intervalSeconds 发送间隔（秒）
     */
    public void startMonitorDataSender(String registerKey, String appName, int intervalSeconds) {
        if (threadPoolMonitor == null) {
            log.warn("线程池监控器未设置，无法启动监控数据发送");
            return;
        }

        log.info("启动线程池监控数据发送，间隔：{}秒", intervalSeconds);
        
        monitorScheduler = Executors.newSingleThreadScheduledExecutor(r -> {
            Thread thread = new Thread(r, "thread-pool-monitor-sender");
            thread.setDaemon(true);
            return thread;
        });

        monitorScheduler.scheduleAtFixedRate(() -> {
            try {
                // 收集监控数据
                ThreadPoolMonitorDTO monitorData = threadPoolMonitor.collectMonitorData(registerKey, appName);
                
                // 发送监控数据到调度器
                jobExecutor.sendMonitorData(monitorData);
                
                log.info("发送线程池监控数据：{}", monitorData);
            } catch (Exception e) {
                log.error("发送线程池监控数据失败", e);
            }
        }, intervalSeconds, intervalSeconds, TimeUnit.SECONDS);
    }

    /**
     * 停止监控数据发送
     */
    public void stopMonitorDataSender() {
        if (monitorScheduler != null) {
            monitorScheduler.shutdown();
            try {
                if (!monitorScheduler.awaitTermination(5, TimeUnit.SECONDS)) {
                    monitorScheduler.shutdownNow();
                }
            } catch (InterruptedException e) {
                monitorScheduler.shutdownNow();
                Thread.currentThread().interrupt();
            }
            log.info("线程池监控数据发送已停止");
        }
    }

    /**
     * 提交任务执行
     * <p>
     * 根据任务的triggerTime和triggerNow字段决定是立即执行还是延迟执行：
     * - 如果triggerNow为true，立即执行任务
     * - 如果triggerTime不为null且大于当前时间，将任务放入时间轮中延迟执行
     * - 如果triggerTime为null或小于等于当前时间，立即执行任务
     *
     * @param request 任务参数
     * @return 任务ID
     */
    public Long submitTask(TaskExecuteRequestDTO request) {
        // 判断是否需要延迟执行
        if (shouldDelayExecute(request)) {
            return submitDelayTask(request);
        } else {
            return submitImmediateTask(request);
        }
    }

    /**
     * 判断是否需要延迟执行
     *
     * @param request 任务请求
     * @return true表示需要延迟执行，false表示立即执行
     */
    private boolean shouldDelayExecute(TaskExecuteRequestDTO request) {
        // 如果triggerNow为true，立即执行
        if (request.isTriggerNow()) {
            return false;
        }

        // 如果triggerTime为null，立即执行
        if (request.getTriggerTime() == null) {
            return false;
        }

        // 计算延迟时间
        long delayMillis = request.getTriggerTime().getTime() - System.currentTimeMillis();

        // 如果延迟时间小于等于0，立即执行
        if (delayMillis <= 0) {
            log.info("任务延迟时间已过期，立即执行: taskId={}, jobName={}, triggerTime={} ,now:{}",
                    request.getJobLogId(), request.getJobName(), DateUtil.format(request.getTriggerTime(), "yyyy-MM-dd HH:mm:ss"),
                    DateUtil.format(new Date(), "yyyy-MM-dd HH:mm:ss"));
            return false;
        }

        return true;
    }

    /**
     * 提交延迟任务
     *
     * @param request 任务请求
     * @return 任务ID
     */
    private Long submitDelayTask(TaskExecuteRequestDTO request) {
        Long taskId = request.getJobLogId();
        String jobName = request.getJobName();

        // 计算延迟时间
        long delayMillis = request.getTriggerTime().getTime() - System.currentTimeMillis();
        long delaySeconds = TimeUnit.MILLISECONDS.toSeconds(delayMillis);

        log.info("提交延迟任务: taskId={}, jobName={}, triggerTime={}, 延迟{}秒",
                taskId, jobName, request.getTriggerTime(), delaySeconds);

        // 创建时间轮任务
        TimerTask timerTask = new TimerTask() {
            @Override
            public void run(Timeout timeout) throws Exception {
                try {
                    log.info("延迟任务时间到，开始执行: taskId={}, jobName={}", taskId, jobName);

                    // 从延迟任务映射表中移除
                    delayTaskMap.remove(taskId);

                    // 提交到线程池立即执行
                    submitImmediateTask(request);

                } catch (Exception e) {
                    log.error("延迟任务执行异常: taskId={}, jobName={}", taskId, jobName, e);
                }
            }
        };

        // 将任务添加到时间轮
        Timeout timeout = delayTimer.newTimeout(timerTask, delayMillis, TimeUnit.MILLISECONDS);

        // 将延迟任务添加到映射表
        delayTaskMap.put(taskId, timeout);

        return taskId;
    }

    /**
     * 提交立即执行的任务
     *
     * @param request 任务请求
     * @return 任务ID
     */
    private Long submitImmediateTask(TaskExecuteRequestDTO request) {
        Long taskId = request.getJobLogId();
        String jobName = request.getJobName();

        FutureTask<TaskExecuteResponseDTO> futureTask = new FutureTask<>(() -> {
            TaskExecuteResponseDTO result = new TaskExecuteResponseDTO();
            result.setJobId(request.getJobId());
            result.setJobLogId(request.getJobLogId());
            result.setJobName(jobName);
            result.setRegisterKey(request.getRegisterKey());
            Date startTime = new Date();
            long startTimeMs = System.currentTimeMillis();
            boolean success = false;
            
            try {
                log.info("开始执行任务: taskId={}, jobName={}", taskId, jobName);
                JobExecutorResultDTO taskExecuteResultDTO = jobExecutor.executeTask(request);
                result.setStatus(JobRunStatusEnum.COMPLETED.getCode());
                result.setResult(taskExecuteResultDTO.getResult());
                success = true;
                log.info("任务执行完成: taskId={}, jobName={}, result={}", taskId, jobName, taskExecuteResultDTO.getResult());
            } catch (Exception e) {
                log.error("任务执行异常: taskId={}, jobName={}", taskId, jobName, e);
                result.setStatus(JobRunStatusEnum.FAILED.getCode());
                result.setResult("执行失败: " + e.getMessage());
                success = false;
            } finally {
                // 任务完成后从映射表中移除
                taskMap.remove(taskId);
                
                // 记录任务执行统计
                if (threadPoolMonitor != null) {
                    long executionTimeMs = System.currentTimeMillis() - startTimeMs;
                    threadPoolMonitor.recordTaskCompletion(executionTimeMs, success);
                }
            }
            Date endTime = new Date();
            result.setStartTime(startTime);
            result.setEndTime(endTime);
            // 执行完成开始发送完成请求
            jobExecutor.sendTaskResult(result);
            return result;
        });

        // 将任务添加到映射表
        taskMap.put(taskId, futureTask);

        // 提交到线程池执行
        taskExecutor.submit(futureTask);

        log.info("任务已提交: taskId={}, jobName={}", taskId, jobName);
        return taskId;
    }

    /**
     * 终止任务
     * <p>
     * 支持终止正在执行的任务和延迟任务：
     * - 如果任务正在执行，取消FutureTask
     * - 如果任务是延迟任务，取消时间轮中的任务
     *
     * @param taskId 任务ID
     * @return 是否成功终止
     */
    public boolean cancelTask(Long taskId) {
        // 首先尝试取消延迟任务
        Timeout delayTimeout = delayTaskMap.get(taskId);
        if (delayTimeout != null) {
            boolean cancelled = delayTimeout.cancel();
            if (cancelled) {
                delayTaskMap.remove(taskId);
                log.info("延迟任务已取消: taskId={}", taskId);
                return true;
            }
        }

        // 尝试取消正在执行的任务
        FutureTask<TaskExecuteResponseDTO> futureTask = taskMap.get(taskId);
        if (futureTask == null) {
            log.warn("任务不存在，无法终止: taskId={}", taskId);
            return false;
        }

        boolean cancelled = futureTask.cancel(true);
        if (cancelled) {
            taskMap.remove(taskId);
            log.info("任务已终止: taskId={}", taskId);
        } else {
            log.warn("任务终止失败: taskId={}", taskId);
        }

        return cancelled;
    }

    /**
     * 获取任务状态
     *
     * @param taskId 任务ID
     * @return 任务状态
     */
    public TaskStatus getTaskStatus(String taskId) {
        // 检查是否是延迟任务
        if (delayTaskMap.containsKey(Long.valueOf(taskId))) {
            return TaskStatus.DELAYED;
        }

        FutureTask<TaskExecuteResponseDTO> futureTask = taskMap.get(Long.valueOf(taskId));
        if (futureTask == null) {
            return TaskStatus.NOT_FOUND;
        }

        if (futureTask.isCancelled()) {
            return TaskStatus.CANCELLED;
        }

        if (futureTask.isDone()) {
            return TaskStatus.COMPLETED;
        }

        return TaskStatus.RUNNING;
    }

    /**
     * 获取任务结果
     *
     * @param taskId  任务ID
     * @param timeout 超时时间
     * @param unit    时间单位
     * @return 任务结果
     */
    public TaskExecuteResponseDTO getTaskResult(String taskId, long timeout, TimeUnit unit) {
        FutureTask<TaskExecuteResponseDTO> futureTask = taskMap.get(Long.valueOf(taskId));
        if (futureTask == null) {
            log.warn("任务不存在: taskId={}", taskId);
            return null;
        }

        try {
            return futureTask.get(timeout, unit);
        } catch (TimeoutException e) {
            log.warn("获取任务结果超时: taskId={}", taskId);
            return null;
        } catch (Exception e) {
            log.error("获取任务结果异常: taskId={}", taskId, e);
            return null;
        }
    }

    /**
     * 关闭任务执行管理器
     */
    public void shutdown() {
        log.info("开始关闭任务执行管理器");

        // 停止监控数据发送
        stopMonitorDataSender();

        // 取消所有延迟任务
        delayTaskMap.forEach((taskId, timeout) -> {
            timeout.cancel();
            log.info("取消延迟任务: taskId={}", taskId);
        });
        delayTaskMap.clear();

        // 取消所有运行中的任务
        taskMap.forEach((taskId, futureTask) -> {
            if (!futureTask.isDone()) {
                futureTask.cancel(true);
                log.info("取消任务: taskId={}", taskId);
            }
        });

        // 清空任务映射表
        taskMap.clear();

        // 关闭时间轮
        delayTimer.stop();

        // 关闭线程池
        taskExecutor.shutdown();
        try {
            if (!taskExecutor.awaitTermination(30, TimeUnit.SECONDS)) {
                taskExecutor.shutdownNow();
            }
        } catch (InterruptedException e) {
            taskExecutor.shutdownNow();
            Thread.currentThread().interrupt();
        }

        log.info("任务执行管理器已关闭");
    }

    /**
     * 生成任务ID
     */
    private String generateTaskId() {
        return "task_" + System.currentTimeMillis() + "_" + taskIdGenerator.incrementAndGet();
    }

    /**
     * 获取线程池状态
     */
    public ThreadPoolStatus getThreadPoolStatus() {
        return new ThreadPoolStatus(taskExecutor.getCorePoolSize(), taskExecutor.getMaximumPoolSize(), taskExecutor.getPoolSize(), taskExecutor.getActiveCount(), taskExecutor.getQueue().size(), taskMap.size(), delayTaskMap.size());
    }

    /**
     * 获取任务执行线程池
     */
    public ThreadPoolExecutor getTaskExecutor() {
        return taskExecutor;
    }

    /**
     * 获取延迟任务数量
     */
    public int getDelayTaskCount() {
        return delayTaskMap.size();
    }

    /**
     * 任务状态枚举
     */
    public enum TaskStatus {
        NOT_FOUND,    // 任务不存在
        RUNNING,      // 运行中
        COMPLETED,    // 已完成
        CANCELLED,    // 已取消
        DELAYED       // 延迟中
    }

    /**
     * 线程池状态
     */
    public static class ThreadPoolStatus {
        private final int corePoolSize;
        private final int maximumPoolSize;
        private final int poolSize;
        private final int activeCount;
        private final int queueSize;
        private final int taskCount;
        private final int delayTaskCount;

        public ThreadPoolStatus(int corePoolSize, int maximumPoolSize, int poolSize, int activeCount, int queueSize, int taskCount, int delayTaskCount) {
            this.corePoolSize = corePoolSize;
            this.maximumPoolSize = maximumPoolSize;
            this.poolSize = poolSize;
            this.activeCount = activeCount;
            this.queueSize = queueSize;
            this.taskCount = taskCount;
            this.delayTaskCount = delayTaskCount;
        }

        // getters
        public int getCorePoolSize() {
            return corePoolSize;
        }

        public int getMaximumPoolSize() {
            return maximumPoolSize;
        }

        public int getPoolSize() {
            return poolSize;
        }

        public int getActiveCount() {
            return activeCount;
        }

        public int getQueueSize() {
            return queueSize;
        }

        public int getTaskCount() {
            return taskCount;
        }

        public int getDelayTaskCount() {
            return delayTaskCount;
        }
    }
} 